package edu.purdue.cs.aggr.message;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;

import edu.purdue.cs.aggr.AggrException;
import edu.purdue.cs.aggr.Node;
import edu.purdue.cs.aggr.messageprocessor.MessageProcessorFactory;

public class MessageListener implements Runnable {
	
	private ServerSocket serverSocket;
	private boolean shutdown = false;
	public final int MAX_EXCEPTION_COUNT = 10;
	
	Node node = null;
	
	long lastMessageReceived = System.currentTimeMillis();
	
	public MessageListener(Node node) {
		this.node = node;
	}
	
	public void handleMessage(Message message) throws AggrException {
//		System.out.println("Got new message...");
		System.out.println("Got new message. Time since last message :" + (System.currentTimeMillis() - lastMessageReceived));
		lastMessageReceived = System.currentTimeMillis();
		MessageProcessorFactory.getMessageProcessor(node, message).processMessage(message);
	}
	
	@Override
	public void run() {
		System.out.println("Starting Aggr server at port : " + node.getPort());
		serverSocket = null;
		
		try {
			serverSocket = new ServerSocket(node.getPort());
		} catch (IOException e1) {
			System.out.println("ERROR: Port taken trying next one...");
			try {
				serverSocket = new ServerSocket(node.getPort()+1);
			} catch (IOException e) {
				e.printStackTrace();
				System.out.println("ERROR: Listener exitting since both ports are taken...");
			}
		}
		
		try {
			while (!shutdown) {
				try {

//					System.out.println("Aggr Server waiting 2... ");
					Socket socket = serverSocket.accept();
					System.out.println("Aggr Server accepted ... ");

					msgCount++;
					long start = System.currentTimeMillis();
					if (msgCount == 2) {
						System.out.println("First message received...");
						firstAccMessageReceived = start;
					} else if (firstAccMessageReceived > 0){
//						System.out.println("Message Interval 1 :" + (start - firstAccMessageReceived));
					}
					
					new Thread(new MessageListenerWorker(socket)).start();
					System.out.println("New Aggr conenction accepted ... ");

				} catch (IOException e) {
					e.printStackTrace(System.out);
				}
			}
			System.out.println("Shutdown state detected.");
		} finally {
			System.out.println("Server exiting ... ");
		}
		
	}
	
	private static long firstAccMessageReceived = -1;
	private static int msgCount = 0;
	
	class MessageListenerWorker implements Runnable {
		private Socket socket = null;
		private ObjectInputStream input = null;
		private ObjectOutputStream output = null;

		
		public MessageListenerWorker (Socket socket) {
			this.socket = socket;
		}
		
		@Override
		public void run() {
			long start = System.currentTimeMillis();
			try {
				input = new ObjectInputStream(socket.getInputStream());
				output = new ObjectOutputStream(socket.getOutputStream());
				boolean keepRunning = true;
				int exceptionCount = 0;
				
				while(keepRunning) {
					try {
						Message message = getNextMessage();
						if(message == null) {
	            System.out.println("Detected that the socket was closed from the other side. Exitting");
						  keepRunning = false;
						} else {
						  handleMessage (message);
						}
					} catch (AggrException e) {
						e.printStackTrace();
						exceptionCount++;
					} catch (java.net.SocketException se) {
						System.out.println("Detected that the socket was closed from the other side. Exitting");
						keepRunning = false;
					}
					
					if(exceptionCount >= MAX_EXCEPTION_COUNT) {
					    System.out.println("Exitting listener since MAX_EXCEPTION_COUNT is reached ...");
					    keepRunning = false;
					}
				}
			} catch (Exception e) {
				e.printStackTrace();
			} 
			long end = System.currentTimeMillis();
//			System.out.println("Message Read Time:" + (end-start));
		}
		
		private Message getNextMessage() throws AggrException, java.net.SocketException {
			try {
			    Object obj = input.readObject();
				
				if(obj instanceof java.net.SocketException) {
					throw (java.net.SocketException)obj;
				}
				
				if(!(obj instanceof Message))
					throw new AggrException("Incompatible message types in the stream");
				
				return (Message) obj;
			} catch (SocketException e) {
			  return null;
			}	catch (IOException e) {
				e.printStackTrace();
			} catch (ClassNotFoundException e) {
				e.printStackTrace();
			}
			
			return null;
		}
		
	}
	
	
	

}
